package com.asp.bigdata.flink.kafka2hdfs.utils;

import com.alibaba.fastjson.JSON;
import com.asp.bigdata.flink.kafka2hdfs.pojo.Event;
import com.zyhlw.DataEncrypt.des.CipherPBEMd5Des;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.io.Serializable;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.Map;
import java.util.Properties;
import java.util.TimeZone;

public class DealUtils implements Serializable {
    private static final String PWD = "zxc@._875";

    public SingleOutputStreamOperator<Event> dealCsv(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+08:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    long time;
                    try {
                        String regexStr = jobConfig.getProperty("split.regex");
                        String[] messgeArr = value.split(regexStr);

                        String even_time = messgeArr[Integer.valueOf(jobConfig.getProperty("time.index"))];
                        time = simpleDateFormat.parse(even_time).getTime();

                        String desdata_index = jobConfig.getProperty("desdata.index");

                        // 是否加密
                        if (desdata_index != null && jobConfig.getProperty("des.pwd") != null) {
                            String[] desdataArr = desdata_index.split(",");

                            for (String des_index : desdataArr) {
                                String pre_data = messgeArr[Integer.valueOf(des_index)];

                                if (!pre_data.equals("") && !pre_data.equals("NULL") && !pre_data.equals("-") && !pre_data.equals("null")) {
                                    if (des_index != "0" || des_index != String.valueOf(messgeArr.length)) {
                                        String realPwd = CipherPBEMd5Des.decrypt(jobConfig.getProperty("des.pwd"), PWD);

                                        String des_data = CipherPBEMd5Des.encrypt(pre_data, realPwd);

                                        StringBuffer preBuffer = new StringBuffer(pre_data).append(regexStr).insert(0, regexStr);
                                        StringBuffer desBuffer = new StringBuffer(des_data).append(regexStr).insert(0, regexStr);

                                        value = value.replaceFirst(preBuffer.toString(), desBuffer.toString());
                                    } else {
                                        throw new Exception("fail to encrypting data！reason：加密字段在首尾");
                                    }
                                }
                            }
                        }

                    } catch (Exception e) {
                        e.printStackTrace();
                        time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                    }

                    out.collect(new Event(value, time, null));
                });
    }

    public SingleOutputStreamOperator<Event> dealJson(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+00:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);
                    long time;

                    String type = jobConfig.getProperty("topic");

                    try {
                        String even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                        time = simpleDateFormat.parse(even_time).getTime();

                    } catch (Exception e) {
                        e.printStackTrace();
                        time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                    }

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    // 多对一，type合并
                    String mergeType = mergeType(jobConfig, type);
                    if (mergeType != null) {
                        type = mergeType;
                    }

                    // needn't decrypt types  time
                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {

                                value = JSON.toJSONString(valueMap);
                                out.collect(new Event(value, time, typeDe));
                                break;
                            }
                        }
                    }

                    String desPwd = jobConfig.getProperty("des.pwd");

                    // 添加支持当topic里有多个type时，能指定type入库到hdfs
                    String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");

                    if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                        // 有decryptType就指定输出，没有直接topic输出
                        Map<String, String> fieldMap = (Map) JSON.parse(needDecryptTypes);
                        String needDecryptType, needDecryntFields;

                        for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {
                            needDecryptType = str2Map.getKey();

                            if (needDecryptType.equals(type)) {

                                needDecryntFields = str2Map.getValue();
                                for (String desField : needDecryntFields.split(",")) {
                                    if (valueMap.get(desField) != null) {
                                        String realPwd = CipherPBEMd5Des.decrypt(desPwd, PWD);
                                        String encrypt_data = CipherPBEMd5Des.encrypt(valueMap.get(desField).toString(), realPwd);
                                        valueMap.put(desField, encrypt_data);
                                    }
                                }

                                value = JSON.toJSONString(valueMap);
                                out.collect(new Event(value, time, needDecryptType));
                                break;
                            }

                        }
                    } else if (notDecryptTypes == null && needDecryptTypes == null) {
                        value = JSON.toJSONString(valueMap);
                        out.collect(new Event(value, time, type));
                    }

                });
    }

    public SingleOutputStreamOperator<Event> dealJson2Csv(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+00:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);
                    long time;

                    String type = jobConfig.getProperty("topic");

                    try {
                        String even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                        time = simpleDateFormat.parse(even_time).getTime();
                    } catch (Exception e) {
                        e.printStackTrace();
                        time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                    }

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    // 多对一，type合并
                    String mergeType = mergeType(jobConfig, type);
                    if (mergeType != null) {
                        type = mergeType;
                    }

                    // needn't decrypt types  time
                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {

                                out.collect(new Event(StringUtils.join(valueMap.values(), "|"), time, typeDe));
                            }
                        }
                    }

                    String desPwd = jobConfig.getProperty("des.pwd");

                    // 添加支持当topic里有多个type时，能指定type入库到hdfs
                    String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");

                    if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                        // 有decryptType就指定输出，没有直接topic输出
                        Map<String, String> fieldMap = (Map) JSON.parse(needDecryptTypes);
                        String needDecryptType, needDecryntFields;

                        for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {
                            needDecryptType = str2Map.getKey();

                            if (needDecryptType.equals(type)) {
                                needDecryntFields = str2Map.getValue();
                                for (String desField : needDecryntFields.split(",")) {
                                    if (valueMap.get(desField) != null) {
                                        String realPwd = CipherPBEMd5Des.decrypt(desPwd, PWD);
                                        String encrypt_data = CipherPBEMd5Des.encrypt(valueMap.get(desField).toString(), realPwd);
                                        valueMap.put(desField, encrypt_data);
                                    }
                                }

                                out.collect(new Event(StringUtils.join(valueMap.values(), "|"), time, needDecryptType));
                                break;
                            }
                        }
                    } else if (notDecryptTypes == null && needDecryptTypes == null) {

                        out.collect(new Event(StringUtils.join(valueMap.values(), "|"), time, type));
                    }
                });
    }

    public SingleOutputStreamOperator<Event> dealJsonMessage(DataStreamSource<String> streamSource, Properties jobConfig) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone(jobConfig.getProperty("time.zone", "GMT+00:00")));
        return streamSource.flatMap(
                (FlatMapFunction<String, Event>) (value, out) -> {
                    Map<String, String> valueMap = (Map) JSON.parse(value);

                    long time;
                    String message = "", even_time = "";
                    String regexStr = jobConfig.getProperty("split.regex");
                    String type = jobConfig.getProperty("topic");

                    // 获取type字段用于分区(表)，如果没有则属于脏数据，使用topic代替
                    type = valueMap.getOrDefault("type", type);
                    // 多对一，type合并
                    String mergeType = mergeType(jobConfig, type);
                    if (mergeType != null) {
                        type = mergeType;
                    }

                    String desPwd = jobConfig.getProperty("des.pwd");
                    String needDecryptTypes = jobConfig.getProperty("needDecryptTypes");
                    message = valueMap.get("message");

                    String[] valueArr = message.split(regexStr);

                    // needn't decrypt types time
                    String notDecryptTypes = jobConfig.getProperty("notDecryptTypes");
                    if (notDecryptTypes != null && !notDecryptTypes.isEmpty()) {
                        for (String typeDe : notDecryptTypes.split(",")) {
                            if (type.equals(typeDe)) {

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                            }
                        }
                    }


                    // need decrypt types 有decryptType就指定输出，没有直接topic输出
                    if (needDecryptTypes != null && !needDecryptTypes.isEmpty() && desPwd != null) {
                        // 有decryptType就指定输出，没有直接topic输出
                        Map<String, String> fieldMap = (Map) JSON.parse(needDecryptTypes);
                        String needDecryptType, needDecryntIndex;

                        for (Map.Entry<String, String> str2Map : fieldMap.entrySet()) {
                            needDecryptType = str2Map.getKey();

                            if (needDecryptType.equals(type)) {
                                needDecryntIndex = str2Map.getValue();

                                for (String des_index : needDecryntIndex.split(",")) {
                                    String pre_data = valueArr[Integer.valueOf(des_index)];
                                    if (!pre_data.equals("") && !pre_data.equals("NULL") && !pre_data.equals("-") && !pre_data.equals("null")) {
                                        if (des_index != "0" || des_index != String.valueOf(valueArr.length)) {
                                            String realPwd = CipherPBEMd5Des.decrypt(desPwd, PWD);

                                            String des_data = CipherPBEMd5Des.encrypt(pre_data, realPwd);

                                            StringBuffer preBuffer = new StringBuffer(pre_data).append(regexStr).insert(0, regexStr);
                                            StringBuffer desBuffer = new StringBuffer(des_data).append(regexStr).insert(0, regexStr);

                                            message = message.replaceFirst(preBuffer.toString(), desBuffer.toString());
                                        } else {
                                            throw new Exception("fail to encrypting data！reason：加密字段在首尾");
                                        }
                                    }
                                }

                                try {
                                    if (jobConfig.getProperty("time.field.index") != null) {
                                        even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                                    } else {
                                        even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                                    }
                                    time = simpleDateFormat.parse(even_time).getTime();

                                } catch (Exception e) {
                                    e.printStackTrace();
                                    time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                                }

                                out.collect(new Event(message, time, type));
                                break;
                            }
                        }
                    } else if (notDecryptTypes == null && needDecryptTypes == null) {

                        try {
                            if (jobConfig.getProperty("time.field.index") != null) {
                                even_time = valueArr[Integer.valueOf(jobConfig.getProperty("time.field.index"))];
                            } else {
                                even_time = valueMap.get(jobConfig.getProperty("time.field", "@timestamp"));
                            }
                            time = simpleDateFormat.parse(even_time).getTime();

                        } catch (Exception e) {
                            e.printStackTrace();
                            time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
                        }

                        out.collect(new Event(message, time, type));
                    }

                });
    }

    private String mergeType(Properties jobConfig, String type) {
        String mergeType = jobConfig.getProperty("mergeType");
        if (mergeType != null) {
            String[] split = mergeType.split("\\|");
            for (String typeMer : split[0].split(",")) {
                if (type.equals(typeMer)) {
                    return split[1];
                }
            }
        }
        return null;
    }

}
